Please address the comments from this code review:

## Overall Comments
- Consider exposing the `min_items_for_parallel` threshold as a constructor parameter (similar to `chunk_size`) rather than hard-coding it to 10 inside `convert_dict_parallel`, so users can tune when dict parallelization kicks in.
- Avoid dynamic imports within the hot path of `dicttoxml`—move `from json2xml.parallel import …` to module level to eliminate repetitive import overhead during conversion.
- Benchmarks show limited gains on large datasets due to string concatenation and pretty-printing overhead; consider profiling those hotspots and using more efficient string builders (e.g. `io.StringIO`) or streaming output to improve throughput beyond the medium-size sweet spot.

## Individual Comments

### Comment 1
<location> `json2xml/dicttoxml.py:726-731` </location>
<code_context>
+
+        if root:
+            output.append('<?xml version="1.0" encoding="UTF-8" ?>')
+            if isinstance(obj, dict):
+                output_elem = convert_dict_parallel(
+                    obj, ids, custom_root, attr_type, item_func, cdata, item_wrap,
+                    list_headers=list_headers, workers=workers, min_items_for_parallel=10
+                )
+            elif isinstance(obj, Sequence):
+                output_elem = convert_list_parallel(
+                    obj, ids, custom_root, attr_type, item_func, cdata, item_wrap,
</code_context>

<issue_to_address>
**issue (bug_risk):** Using Sequence to check for list-like objects may include strings and bytes.

Explicitly exclude str and bytes when checking for Sequence to prevent unintended handling of these types.
</issue_to_address>

### Comment 2
<location> `json2xml/parallel.py:14-21` </location>
<code_context>
+```python
+import sys
+
+def is_free_threaded() -> bool:
+    """Check if running on free-threaded Python build."""
+    return hasattr(sys, '_is_gil_enabled') and not sys._is_gil_enabled()
</code_context>

<issue_to_address>
**suggestion:** Reliance on sys._is_gil_enabled() may break in future Python versions.

Document the use of this private attribute and consider adding a fallback to handle potential changes in future Python versions.

```suggestion
def is_free_threaded() -> bool:
    """
    Check if running on free-threaded Python build (Python 3.13t).

    Note:
        This function relies on the private attribute `sys._is_gil_enabled`, which may change or be removed in future Python versions.
        If the attribute is not present, or its semantics change, this function will fall back to assuming GIL is enabled.

    Returns:
        bool: True if running on free-threaded build, False otherwise.
    """
    # Fallback: If attribute is missing or not callable, assume GIL is enabled.
    gil_enabled = True
    if hasattr(sys, '_is_gil_enabled'):
        try:
            gil_enabled = sys._is_gil_enabled()
        except Exception:
            pass
    return not gil_enabled
```
</issue_to_address>

### Comment 3
<location> `json2xml/parallel.py:24` </location>
<code_context>
+    return hasattr(sys, '_is_gil_enabled') and not sys._is_gil_enabled()
+
+
+def get_optimal_workers(workers: int | None = None) -> int:
+    """
+    Get the optimal number of worker threads.
</code_context>

<issue_to_address>
**suggestion (performance):** Limiting workers to min(4, cpu_count) for non-free-threaded Python may be too restrictive.

Making the worker limit configurable or informed by benchmarks could better utilize system resources on machines with many cores.

Suggested implementation:

```python
def get_optimal_workers(
    workers: int | None = None,
    max_workers_limit: int | None = None
) -> int:
    """
    Get the optimal number of worker threads.

    Args:
        workers: Explicitly specified worker count. If None, auto-detect.
        max_workers_limit: Optional cap for worker count on non-free-threaded Python.

    Returns:
        int: Number of worker threads to use.
    """
    if workers is not None:
        return max(1, workers)

```

```python
    if workers is not None:
        return max(1, workers)

    import os

    cpu_count = os.cpu_count() or 1

    if is_free_threaded():
        optimal = cpu_count
    else:
        # Use configurable limit or default to 4
        limit = max_workers_limit if max_workers_limit is not None else 4
        optimal = min(limit, cpu_count)

    return max(1, optimal)

```
</issue_to_address>

### Comment 4
<location> `json2xml/parallel.py:217-226` </location>
<code_context>
+       if len(obj) < 10:
+           return convert_dict(obj, ids, parent, attr_type, item_func, cdata, item_wrap, list_headers)
+       
+       with ThreadPoolExecutor(max_workers=workers) as executor:
+           futures = []
+           for key, val in obj.items():
</code_context>

<issue_to_address>
**suggestion (bug_risk):** Using as_completed may result in out-of-order results.

If a future raises an exception, the current error handling may not be sufficient. Please add error handling for failed futures.
</issue_to_address>

### Comment 5
<location> `tests/test_parallel.py:60-58` </location>
<code_context>
+        )
+        assert result_parallel == result_serial
+
+    def test_parallel_dict_large(self) -> None:
+        """Test parallel dict conversion with large data."""
+        data = {f"key{i}": f"value{i}" for i in range(20)}
+        result_parallel = convert_dict_parallel(
+            data, [], "root", True, dicttoxml.default_item_func, False, True, False, workers=4
+        )
+        result_serial = dicttoxml.convert_dict(
+            data, [], "root", True, dicttoxml.default_item_func, False, True, False
+        )
+        assert result_parallel == result_serial
+
+    def test_parallel_list_small(self) -> None:
</code_context>

<issue_to_address>
**suggestion (testing):** Consider adding tests for error handling in parallel conversion functions.

Please add tests that check how parallel conversion functions handle invalid inputs and ensure appropriate exceptions are raised.

Suggested implementation:

```python
    def test_parallel_dict_invalid_input(self) -> None:
        """Test parallel dict conversion with invalid input types."""
        # Passing a list instead of a dict
        invalid_data = ["not", "a", "dict"]
        with pytest.raises(TypeError):
            convert_dict_parallel(
                invalid_data, [], "root", True, dicttoxml.default_item_func, False, True, False, workers=2
            )

        # Passing None
        with pytest.raises(TypeError):
            convert_dict_parallel(
                None, [], "root", True, dicttoxml.default_item_func, False, True, False, workers=2
            )

    def test_parallel_list_invalid_input(self) -> None:
        """Test parallel list conversion with invalid input types."""
        # Passing a dict instead of a list
        invalid_data = {"not": "a list"}
        with pytest.raises(TypeError):
            convert_list_parallel(
                invalid_data, [], "root", True, dicttoxml.default_item_func, False, True, False, workers=2, chunk_size=100
            )

        # Passing None
        with pytest.raises(TypeError):
            convert_list_parallel(
                None, [], "root", True, dicttoxml.default_item_func, False, True, False, workers=2, chunk_size=100
            )

    def test_parallel_list_small(self) -> None:
        """Test parallel list conversion with small data (should fallback to serial)."""
        data = ["item1", "item2", "item3"]
        result_parallel = convert_list_parallel(
            data, [], "root", True, dicttoxml.default_item_func, False, True, False, workers=2, chunk_size=100
        )
        result_serial = dicttoxml.convert_list(
            data, [], "root", True, dicttoxml.default_item_func, False, True, False
        )
        assert result_parallel == result_serial


```

These tests assume that `convert_dict_parallel` and `convert_list_parallel` raise `TypeError` for invalid input types. If they raise a different exception (e.g., `ValueError`), adjust the `pytest.raises` argument accordingly.

You must also ensure that `pytest` is imported at the top of the file:
```python
import pytest
```
If not already present, add this import.
</issue_to_address>

### Comment 6
<location> `tests/test_parallel.py:202-211` </location>
<code_context>
+
+        assert result_parallel == result_serial
+
+    def test_parallel_with_special_characters(self) -> None:
+        """Test parallel processing with special XML characters."""
+        data = {
+            f"key{i}": f"value with <special> & \"characters\" {i}"
+            for i in range(15)
+        }
+
+        result_parallel = dicttoxml.dicttoxml(data, parallel=True, workers=4)
+        result_serial = dicttoxml.dicttoxml(data, parallel=False)
+
+        assert result_parallel == result_serial
+        assert b"&lt;special&gt;" in result_parallel
+        assert b"&amp;" in result_parallel
+
+    def test_parallel_empty_data(self) -> None:
</code_context>

<issue_to_address>
**suggestion (testing):** Consider adding tests for other XML edge cases, such as attributes, namespaces, and CDATA sections.

Adding tests for attributes, namespaces, and CDATA will help verify that parallel processing handles these XML features correctly and avoids subtle bugs.

Suggested implementation:

```python
    def test_parallel_with_attributes(self) -> None:
        """Test parallel processing with XML attributes."""
        # Simulate attribute handling using dicttoxml's attr_type feature
        data = {
            "person": {
                "@id": "123",
                "name": "Alice"
            }
        }
        result_parallel = dicttoxml.dicttoxml(data, parallel=True, workers=2, attr_type=True)
        result_serial = dicttoxml.dicttoxml(data, parallel=False, attr_type=True)
        assert result_parallel == result_serial
        assert b'id="123"' in result_parallel

    def test_parallel_with_namespaces(self) -> None:
        """Test parallel processing with XML namespaces."""
        # Simulate namespace handling by including a namespace in the tag
        data = {
            "ns:person": {
                "name": "Bob"
            }
        }
        result_parallel = dicttoxml.dicttoxml(data, parallel=True, workers=2)
        result_serial = dicttoxml.dicttoxml(data, parallel=False)
        assert result_parallel == result_serial
        assert b"<ns:person>" in result_parallel

    def test_parallel_with_cdata(self) -> None:
        """Test parallel processing with CDATA sections."""
        # Simulate CDATA by including a value that should be wrapped in CDATA
        data = {
            "note": "<![CDATA[Some <cdata> content & more]]>"
        }
        result_parallel = dicttoxml.dicttoxml(data, parallel=True, workers=2)
        result_serial = dicttoxml.dicttoxml(data, parallel=False)
        assert result_parallel == result_serial
        assert b"<![CDATA[Some <cdata> content & more]]>" in result_parallel

    def test_parallel_empty_data(self) -> None:
        """Test parallel processing with empty data."""
        data = {"key": "value"}
        converter = Json2xml(data, parallel=True, workers=4)
        result = converter.to_xml()
        assert result is not None

```

- If your dicttoxml library or wrapper does not support attributes, namespaces, or CDATA natively, you may need to adjust the test data or parameters to match your implementation.
- Ensure that the dicttoxml.dicttoxml function supports the `attr_type` argument for attributes. If not, you may need to use a different approach or library for these tests.
- If your XML generator does not support CDATA sections, you may need to extend its functionality or mock the expected output.
</issue_to_address>

### Comment 7
<location> `tests/test_parallel.py:216-221` </location>
<code_context>
+        assert b"&lt;special&gt;" in result_parallel
+        assert b"&amp;" in result_parallel
+
+    def test_parallel_empty_data(self) -> None:
+        """Test parallel processing with empty data."""
+        data = {"key": "value"}
+        converter = Json2xml(data, parallel=True, workers=4)
+        result = converter.to_xml()
+        assert result is not None
+
+    def test_parallel_with_none_workers(self) -> None:
</code_context>

<issue_to_address>
**nitpick (testing):** Consider adding assertions for the actual output of empty data conversion.

Asserting the exact output for empty data will make the test more reliable and help detect unintended changes in the output format.
</issue_to_address>

### Comment 8
<location> `tests/test_parallel.py:223-221` </location>
<code_context>
+        result = converter.to_xml()
+        assert result is not None
+
+    def test_parallel_with_none_workers(self) -> None:
+        """Test parallel processing with None workers (auto-detect)."""
+        data = {f"key{i}": f"value{i}" for i in range(20)}
+        converter = Json2xml(data, parallel=True, workers=None)
+        result = converter.to_xml()
+        assert result is not None
+
+    def test_parallel_dict_order_preserved(self) -> None:
</code_context>

<issue_to_address>
**suggestion (testing):** Consider adding a test for chunk_size=None or invalid chunk_size values.

Testing chunk_size=None or invalid values (e.g., negative, zero) will verify that the code handles these scenarios correctly and applies appropriate defaults.
</issue_to_address>

### Comment 9
<location> `json2xml/dicttoxml.py:721` </location>
<code_context>
-        )
-        output.append(f"<{custom_root}{namespace_str}>{output_elem}</{custom_root}>")
+
+    if parallel:
+        from json2xml.parallel import convert_dict_parallel, convert_list_parallel
+
</code_context>

<issue_to_address>
**issue (complexity):** Consider refactoring the conversion logic into a single helper function to centralize dispatch and eliminate nested conditionals.

```python
# add this helper at module‐level to unify dispatching
def _dispatch_convert(
    obj, ids, parent,
    attr_type, item_func, cdata, item_wrap, list_headers,
    parallel, workers, chunk_size
):
    if parallel:
        if isinstance(obj, dict):
            return convert_dict_parallel(
                obj, ids, parent,
                attr_type=attr_type, item_func=item_func, cdata=cdata,
                item_wrap=item_wrap, list_headers=list_headers,
                workers=workers, min_items_for_parallel=10
            )
        if isinstance(obj, Sequence):
            return convert_list_parallel(
                obj, ids, parent,
                attr_type=attr_type, item_func=item_func, cdata=cdata,
                item_wrap=item_wrap, list_headers=list_headers,
                workers=workers, chunk_size=chunk_size
            )
    # fallback to serial
    return convert(
        obj, ids,
        attr_type, item_func, cdata, item_wrap,
        parent=parent, list_headers=list_headers
    )


def dicttoxml(
    obj: ELEMENT,
    root: bool = True,
    custom_root: str = "root",
    ids: list[int] | None = None,
    attr_type: bool = True,
    item_wrap: bool = True,
    item_func: Callable[[str], str] = default_item_func,
    cdata: bool = False,
    xml_namespaces: dict[str, Any] = {},
    list_headers: bool = False,
    parallel: bool = False,
    workers: int | None = None,
    chunk_size: int = 100
) -> bytes:
    # ... build namespace_str ...

    # choose parent name
    parent = custom_root if root else ""
    # single dispatch point
    output_elem = _dispatch_convert(
        obj, ids, parent,
        attr_type, item_func, cdata, item_wrap, list_headers,
        parallel, workers, chunk_size
    )

    if root:
        output = ['<?xml version="1.0" encoding="UTF-8" ?>']
        output.append(f"<{custom_root}{namespace_str}>{output_elem}</{custom_root}>")
    else:
        output = [output_elem]

    return "".join(output).encode("utf-8")
```

This removes the duplicated `if root`/`elif parallel` nesting and centralizes all conversion logic into one flat helper.
</issue_to_address>

### Comment 10
<location> `json2xml/parallel.py:49` </location>
<code_context>
+_validation_cache_lock = threading.Lock()
+_validation_cache: dict[str, bool] = {}
+
+def key_is_valid_xml_cached(key: str) -> bool:
+    """Thread-safe cached version of key_is_valid_xml."""
+    with _validation_cache_lock:
</code_context>

<issue_to_address>
**issue (complexity):** Consider refactoring the parallel functions to use a generic executor wrapper and functools.lru_cache for caching to reduce duplication and thread-safety code.

```suggestion
Consider collapsing the two “parallel” functions into one generic executor wrapper and swapping out the manual cache/lock for `functools.lru_cache`. This removes almost all duplication and thread‐safety boilerplate while preserving behavior.

1) Replace manual cache + lock with `lru_cache`:

```python
from functools import lru_cache

@lru_cache(maxsize=None)
def key_is_valid_xml(key: str) -> bool:
    return dicttoxml.key_is_valid_xml(key)
```

2) Add a single helper to parallelize any function over a list of args:

```python
from concurrent.futures import ThreadPoolExecutor, as_completed

def _parallel_map(
    fn: Callable, 
    args_list: list[tuple], 
    workers: int | None = None, 
    min_items: int = 0
) -> list[Any]:
    if len(args_list) <= min_items:
        return [fn(*args) for args in args_list]
    workers = get_optimal_workers(workers)
    results = [None] * len(args_list)
    with ThreadPoolExecutor(max_workers=workers) as executor:
        future_to_idx = {
            executor.submit(fn, *args): idx
            for idx, args in enumerate(args_list)
        }
        for fut in as_completed(future_to_idx):
            results[future_to_idx[fut]] = fut.result()
    return results
```

3) Refactor your two APIs to use `_parallel_map`. Example for dict:

```python
def convert_dict_parallel(...):
    items = list(obj.items())
    args = [
      (key, val, ids, parent, attr_type, item_func, cdata, item_wrap, list_headers)
      for key, val in items
    ]
    parts = _parallel_map(_convert_dict_item, args, workers, min_items_for_parallel)
    return "".join(parts)
```

And similarly for lists. This removes ~200 lines of duplicated pool setup, cache, and ordering logic.```
</issue_to_address>

### Comment 11
<location> `json2xml/parallel.py:132` </location>
<code_context>
    attr = {} if not ids else {"id": f"{dicttoxml.get_unique_id(parent)}"}

</code_context>

<issue_to_address>
**suggestion (code-quality):** Swap if/else branches of if expression to remove negation ([`swap-if-expression`](https://docs.sourcery.ai/Reference/Rules-and-In-Line-Suggestions/Python/Default-Rules/swap-if-expression))

```suggestion
    attr = {"id": f"{dicttoxml.get_unique_id(parent)}"} if ids else {}
```

<br/><details><summary>Explanation</summary>Negated conditions are more difficult to read than positive ones, so it is best
to avoid them where we can. By swapping the `if` and `else` conditions around we
can invert the condition and make it positive.
</details>
</issue_to_address>

### Comment 12
<location> `benchmark.py:61` </location>
<code_context>
def run_benchmarks():
    """Run comprehensive benchmarks."""
    print("=" * 80)
    print("json2xml Performance Benchmark")
    print("=" * 80)
    print(f"Python Version: {sys.version}")
    print(f"Python Executable: {sys.executable}")
    print(f"Free-threaded: {'Yes' if is_free_threaded() else 'No'}")
    gil_status = "Disabled (Free-threaded)" if is_free_threaded() else "Enabled (Standard GIL)"
    print(f"GIL Status: {gil_status}")
    print("=" * 80)
    print()

    sizes = ["small", "medium", "large", "xlarge"]

    for size in sizes:
        print(f"\n{size.upper()} Dataset:")
        print("-" * 80)

        data = generate_test_data(size)

        # Count items
        if "users" in data:
            item_count = len(data["users"])
        elif "data" in data:
            item_count = len(data["data"])
        else:
            item_count = len(data)
        print(f"Items: {item_count}")

        # Serial benchmark
        serial_time = benchmark_conversion(data, parallel=False)
        print(f"Serial:     {serial_time*1000:.2f} ms")

        # Parallel benchmarks with different worker counts
        for workers in [2, 4, 8]:
            parallel_time = benchmark_conversion(data, parallel=True, workers=workers, chunk_size=100)
            speedup = serial_time / parallel_time
            print(f"Parallel ({workers}w): {parallel_time*1000:.2f} ms (speedup: {speedup:.2f}x)")

        print()

    print("=" * 80)
    print("Benchmark complete!")
    print("=" * 80)

</code_context>

<issue_to_address>
**issue (code-quality):** Extract duplicate code into function ([`extract-duplicate-method`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/extract-duplicate-method/))
</issue_to_address>

### Comment 13
<location> `json2xml/parallel.py:39-42` </location>
<code_context>
def get_optimal_workers(workers: int | None = None) -> int:
    """
    Get the optimal number of worker threads.

    Args:
        workers: Explicitly specified worker count. If None, auto-detect.

    Returns:
        int: Number of worker threads to use.
    """
    if workers is not None:
        return max(1, workers)

    cpu_count = os.cpu_count() or 4

    if is_free_threaded():
        return cpu_count
    else:
        return min(4, cpu_count)

</code_context>

<issue_to_address>
**suggestion (code-quality):** Replace if statement with if expression ([`assign-if-exp`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/assign-if-exp/))

```suggestion
    return cpu_count if is_free_threaded() else min(4, cpu_count)
```
</issue_to_address>